package dataStream;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
import dataStream.Person;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/12
 */
public class DelayWindow {
    public static void main(String[] args) throws Exception {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20, 12345L));
        data.add( new Person("man", 18, 123712L));
        data.add( new Person("man", 16, 12998790L));
        data.add( new Person("woman", 20, 1827961387L));
        data.add( new Person("woman", 18, 12983789L));
        data.add( new Person("woman", 16, 1287637812L));
        data.add( new Person("woman", 15,897126L));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        OutputTag<Person> late = new OutputTag<Person>("late"){};

        DataStream<Person> dataStream = env.fromCollection(data);
        SingleOutputStreamOperator<Person> singleOutputStreamOperator = dataStream.keyBy(person -> person.gender)
                .window(TumblingEventTimeWindows.of(Time.hours(1L)))
                .allowedLateness(Time.seconds(10L))
                .sideOutputLateData(late)
                .sum("age");

        DataStream<Person> sideOutput = singleOutputStreamOperator.getSideOutput(late);
        sideOutput.print();

        singleOutputStreamOperator.print();

        env.execute();
    }
}
